package org.hbhk.hms.consumer;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.hbhk.hms.entity.HmsTopic;
import org.hbhk.hms.mq.MqConsumer;
import org.hbhk.hms.service.IHmsTopicService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;

/**
 * 
 * 消费处理，对应的topic未触发数据
 *
 */
@Service
public class HmsTopicDataConsumer implements InitializingBean {

	private Logger log = LoggerFactory.getLogger(getClass());

	@Value("${mq.subscribe.HmsDelalyData:HmsDelalyData}")
	public String subscribe = "HmsDelalyData";

	@Value("${mq.reconsumeTimes:3}")
	public Integer reconsumeTimes = 3;

	@Value("${mq.consumeMessageBatchMaxSize:1}")
	public Integer consumeMessageBatchMaxSize = 1;

	private DefaultMQPushConsumer consumer;

	@Autowired
	private IHmsTopicService hmsTopicService;

	@Autowired
	private MqConsumer mqConsumer;

	public void init() throws Exception {

		log.error("HmsDelalyData consumer start...");

		consumer = mqConsumer.get("HmsDelayData_consumer");
		consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
		consumer.subscribe(subscribe, "*");

		consumer.registerMessageListener(new MessageListenerConcurrently() {

			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
				int reconsumeTimes = 0;
				try {
					if(consumeMessageBatchMaxSize == 1) {
						MessageExt me = msgs.get(0);
						reconsumeTimes = me.getReconsumeTimes();
						long s = System.currentTimeMillis();
						String topic = me.getTopic();
						String tags = me.getTags();
						String keys = me.getKeys();
						String msgBody = new String(me.getBody());
						log.debug("topic: " + topic + ",tags: " + tags + ",key: " + keys + ",msgBody: " + msgBody);
						HmsTopic hmsTopic = JSON.parseObject(msgBody, HmsTopic.class);
						hmsTopicService.insert(hmsTopic);
						long e = System.currentTimeMillis();
						long t = e - s;
						log.info("execution time :" + t);
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}else {
						long s = System.currentTimeMillis();

						List<HmsTopic> hmsTopics = Lists.newArrayList();
						for (MessageExt me : msgs) {
							reconsumeTimes = me.getReconsumeTimes();
							String topic = me.getTopic();
							String tags = me.getTags();
							String keys = me.getKeys();
							String msgBody = new String(me.getBody());
							log.debug("topic: " + topic + ",tags: " + tags + ",key: " + keys + ",msgBody: " + msgBody);
							HmsTopic hmsTopic = JSON.parseObject(msgBody, HmsTopic.class);
							hmsTopics.add(hmsTopic);
						}
						hmsTopicService.insertBatch(hmsTopics);
						long e = System.currentTimeMillis();
						long t = e - s;
						log.info("batch execution time :" + t +" size:" +hmsTopics.size());
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}
				
				} catch (Exception e) {
					log.error(e.getMessage(), e);
					log.error("reconsumeTimes: " + reconsumeTimes);
					if (reconsumeTimes >= reconsumeTimes) {
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}
					return ConsumeConcurrentlyStatus.RECONSUME_LATER;
				}
			}
		});

		consumer.start();
		log.error("HmsDelalyData consumer end...");

	}

	@Override
	public void afterPropertiesSet() throws Exception {
		init();
	}

}
